package net.bwie.realtime.jtp.job;

import com.alibaba.fastjson.JSON;
import net.bwie.realtime.jtp.Rpfunction.*;
import net.bwie.realtime.jtp.bean.Order;
import net.bwie.realtime.jtp.utils.JdbcUtil;
import net.bwie.realtime.jtp.utils.KafkaUtil;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * 简介说明:
 *
 * @author: LiLi
 * @date: 2025/05/27 14:47:49
 * @version: 1.0
 */
public class JdApiLogJob {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStream<String> order_stream = KafkaUtil.consumerKafka(env, "dwd-phone-orders");

        // 转换数据流并设置时间特性
        DataStream<Order> orderDataStream = order_stream
                .map(json -> JSON.parseObject(json, Order.class))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Order>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                .withTimestampAssigner((order, timestamp) -> order.getTimestamp())
                );


        handle(orderDataStream);

        abnormalPatternProcess(orderDataStream);

        env.execute("Phone_Orders_handle");
    }



    private static void handle(DataStream<Order> orderStream) {

//        a. 实时GMV（总成交金额）
        SingleOutputStreamOperator<String> totalSalesWithTime = orderStream
                .keyBy(value -> "total")
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                .process(new JdApiTotalSalesFunction())
                .name("Total Sales Aggregation with Window Time");

        JdbcUtil.sinkToClickhouseUpsert(totalSalesWithTime,
                "INSERT INTO  jd_api_report.dws_total_sales_window_report(\n" +
                        "    window_start_time, " +
                        "    window_end_time,\n" +
                        "    total_sales ,\n" +
                        "    ts\n" +
                        ")\n" +
                        "VALUES (?, ?, ?, ?);"
        );

//        b. 实时订单量
        SingleOutputStreamOperator<String> orderCountAggregation = orderStream
                .map(order -> 1L)
                .keyBy(value -> "all")
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                .aggregate(new OrderCountAggregation(), new OrderCountWindowFunction())
                .name("order count Aggregation with Window Time");

        //        totalSalesWithTime.print().setParallelism(1);

        JdbcUtil.sinkToClickhouseUpsert(orderCountAggregation,
                "INSERT INTO jd_api_report.dws_order_count_window_report(\n" +
                        "    window_start_time, window_end_time,\n" +
                        "    order_count,\n" +
                        "    ts)\n" +
                        "VALUES (?, ?, ?,?)"
        );



//        C．用户参与数

        SingleOutputStreamOperator<String> userCountAggregate = orderStream.map(new MapFunction<Order, Tuple2<Integer, Long>>() {
                    @Override
                    public Tuple2<Integer, Long> map(Order value) throws Exception {
                        return Tuple2.of(value.getUserId(), 1L);
                    }
                }).keyBy(x -> x.f0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                .aggregate(new UserCountAggregateFunction(), new UserCountWindowFunction())
                .name("user Count Aggregate");

//        OrderCountAggregate.print().setParallelism(1);

            JdbcUtil.sinkToClickhouseUpsert(userCountAggregate,
                    "INSERT INTO jd_api_report.dws_user_count_window_report(\n" +
                            "    window_start_time, window_end_time,\n" +
                            "    user_id,user_count,\n" +
                            "    ts)\n" +
                            "VALUES (?, ?, ?,?, ?);"
            );

//        d：热门商品TOP10

        // 5 分钟 开一次窗口
        SingleOutputStreamOperator<String> topNHotItemsAggregation =
                orderStream
                        .keyBy(value -> value.getItemId())
                        .window(TumblingProcessingTimeWindows.of(Time.minutes(5)))
                        .aggregate(new TopNHotAggregation(), new TopNHotWindowFunction())
                        .keyBy(t -> t.f0)
                        .process(new TopNHotStateFunction(10))
                        .name("order count Aggregation with Window Time");

        topNHotItemsAggregation.print().setParallelism(1);
        totalSalesWithTime.print().setParallelism(1);

        JdbcUtil.sinkToClickhouseUpsert(topNHotItemsAggregation,
                "INSERT INTO jd_api_report.dws_topN_item_window_report(\n" +
                        "    window_start_time, window_end_time,\n" +
                        "    item,item_size,\n" +
                        "    ts)\n" +
                        "VALUES (?, ?, ?, ?, ?)");



//        e：地区消费分布
        SingleOutputStreamOperator<String> provinceSalesAggregation = orderStream.keyBy(order -> order.getProvince())
                .window(TumblingEventTimeWindows.of(Time.minutes(10)))
                .aggregate( new ProvincePaymentWindowFunction(),new ProvincePaymentAggregation())
                .name("Province Sales Distribution");

        provinceSalesAggregation.print().setParallelism(1);
        JdbcUtil.sinkToClickhouseUpsert(provinceSalesAggregation,
                "INSERT INTO jd_api_report.dws_province_total_payment_window_report(\n" +
                        "    window_start_time, window_end_time,\n" +
                        "    province,\n" +
                        "    total_payment,\n" +
                        "    ts)\n" +
                        "VALUES (?, ?, ?,?,?);");

//      f．促销活动效果
        SingleOutputStreamOperator<String> promotionEffectAggregation = orderStream
                .filter(order -> order.getPromotionId() != null)
                .keyBy(order -> order.getPromotionId())
                .window(TumblingEventTimeWindows.of(Time.hours(1)))
                .aggregate(new PromotionAggregation(), new PromotionWindowFunction())
                .name("Promotion Effect");


        JdbcUtil.sinkToClickhouseUpsert(promotionEffectAggregation,
                "INSERT INTO jd_api_report.dws_promotion_window_report(\n" +
                        "    window_start_time, window_end_time,\n" +
                        "    promotionId,\n" +
                        "        promotion_payment,\n" +
                        "        promotion_count,\n" +
                        "    ts)\n" +
                        "VALUES (?, ?, ?,?,?,?);");

    }

    //        g．异常交易监控
    private static void abnormalPatternProcess(DataStream<Order> orderDataStream) {
        // 定义阈值常量
        final double largeAmountThreshold = 50000.0;

        // 定义异常交易模式
        Pattern<Order, ?> abnormalPattern = Pattern.<Order>begin("first")
                .where(new SimpleCondition<Order>() {
                    @Override
                    public boolean filter(Order value) {
                        // 检测大额订单
                        return value.getPayment() >= largeAmountThreshold;
                    }
                })
                .followedBy("second")
                .where(new SimpleCondition<Order>() {
                    @Override
                    public boolean filter(Order value) {
                        // 同一用户短时间内再次大额订单
                        return value.getPayment() >= largeAmountThreshold * 0.5;
                    }
                })
                .within(Time.minutes(5));

        // 创建CEP模式流
        PatternStream<Order> patternStream = CEP.pattern(
                orderDataStream.keyBy(order -> order.getUserId()),
                abnormalPattern
        );

        // 应用报警处理器
        DataStream<String> alerts = patternStream.process(
                new AbnormalOrderAlert(5, 50000.0, 0.2)
        );

//         输出报警信息（实际应用中可写入Kafka/邮件等）
        alerts.print().setParallelism(1);
    }

    private static class JdApiTotalSalesFunction extends ProcessWindowFunction<Order, String, String, TimeWindow> {

        private FastDateFormat format = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");

        @Override
        public void process(String string, Context context, Iterable<Order> elements, Collector<String> out) throws Exception {
            TimeWindow window = context.window();

            String windowStartTime = format.format(window.getStart());
            String windowEndTime = format.format(window.getEnd());
            // 计算窗口内总销售额
            double sum = 0.0;
            for (Order element : elements) {
                sum += (element.getQuantity() * element.getPrice());
            }
            // 获取窗口信息

            // 3.输出
            String output = windowStartTime + "," + windowEndTime + ","  + String.format("%.8E", sum) +","+
                    System.currentTimeMillis();
            // 收集结果 窗口开始时间，结束时间 ，总销售额,系统时间
            out.collect(output);
        }
    }
}
